package com.yangfan.thread.thread_wangwenjun04;


import java.util.*;
import java.util.stream.IntStream;

/**
 * @description: 初级版的简单的线程池案例
 * @author: yangfan
 * @date: Created in 2020/1/28 0028 20:11
 * @modified By:
 */
public class SimpleThreadPool extends Thread {

    //线程池大小
    private int size;

    //线程池队列大小
    private final int queueSize;

    private static final int DEFAULT_QUEUE_SIZE = 20;

    private static final LinkedList<Runnable> TASK_QUEUE = new LinkedList<>();//任务队列

    private static final List<WorkerTask> THREAD_QUEUE = new ArrayList<>();

    //拒绝策略接口
    private final RejectionStrategy rejectionStrategy;

    //拒绝策略默认实现，可以在里面抛异常
    private static final RejectionStrategy DEFAULT_REJECTION_STRATEGY = () -> {
        throw new RuntimeException("你不重新定义我我就抛异常啦啦啦啦!");
    };

    //是否关闭默认为否
    private volatile boolean destroy = false;

    //默认的线程组
    private static final ThreadGroup GROUP = new ThreadGroup("SimpleGroup");

    //默认的线程名前缀
    private static final String THREAD_PREFIX = "SimpleThread-";

    //线程名后缀
    private static volatile int sqe = 0;

    private int min;

    private int max;

    private int active;

    private static final int MIN = 4;

    private static final int ACTIVE = 8;

    private static final int MAX = 12;

    //线程的状态
    private enum TaskState {
        FREE, RUNNING, BLOCKED, DEAD
    }

    /**
     * 策略模式 函数式接口
     */
    @FunctionalInterface
    private interface RejectionStrategy {

        void RejectionStrategy() throws RejectionStrategyException;

    }


    //自定义异常 如果没有定义拒绝策略就抛异常
    private class RejectionStrategyException extends RuntimeException {

        RejectionStrategyException(String message) {
            super(message);
        }

    }


    public SimpleThreadPool() {
        this(MIN, ACTIVE, MAX, DEFAULT_QUEUE_SIZE, DEFAULT_REJECTION_STRATEGY);
    }


    public SimpleThreadPool(int min, int active, int max, int queueSize, RejectionStrategy rejectionStrategy) {
        this.min = min;
        this.active = active;
        this.max = max;
        this.queueSize = queueSize;
        this.rejectionStrategy = rejectionStrategy;
        init();
    }


    private void init() {

        //初始化的时候只默认只初始化最小的线程数
        for (int i = 0; i < min; i++) {

            createWorkerTask();

        }
        size = min;
        this.start();

    }


    @Override
    public void run() {

        while (!destroy) {//如果线程池没有被关闭则 进行监控

            try {
                Thread.sleep(2_000);
            } catch (InterruptedException e) {
                e.printStackTrace();
            }

            System.out.printf("min:%d , active:%d , max:%d ,current:%d queueSize:%d \n", min, active, max, size,TASK_QUEUE.size());

            //如果任务数量 大于 当前动态的数量
            //并且 当前线程数量 小于 active
            if (TASK_QUEUE.size() > active && size < active) {

                for (int i = size; i < active; i++) {
                    createWorkerTask();//创建线程
                }

                size = active;
                System.out.println(" =============== 线程扩充到" + size + " =============== ");

            } else if (TASK_QUEUE.size() > max && size < max) {

                for (int i = max; i < max; i++) {
                    createWorkerTask();
                }

                size = max;
                System.out.println(" =============== 线程扩充到" + size + " =============== ");

            }


            //如果没有任务了,则把线程池缩成动态大小
            synchronized (THREAD_QUEUE){
                if (TASK_QUEUE.isEmpty()) {

                    System.out.println("已经没有任务了！！！线程池要缩减成 动态大小");

                    synchronized (TASK_QUEUE) {

                        //计算要释放线程的数量
                        int release = size - active;

                        for (Iterator<WorkerTask> it = THREAD_QUEUE.iterator(); it.hasNext(); ) {

                            if (release <= 0)
                                break;

                            WorkerTask wt = it.next();
                            wt.interrupt();//打断
                            wt.close();//关闭
                            it.remove();//删除线程队列里的
                            release -- ;

                        }
                        size = active;

                    }

                }
            }

        }

    }

    /**
     * 提供外界调用的提交任务的方法
     *
     * @param runnable
     */
    public void submit(Runnable runnable) throws IllegalAccessException {

        if (destroy)
            throw new IllegalStateException("非法的状态!");

        synchronized (TASK_QUEUE) {

            if (TASK_QUEUE.size() > queueSize)
                //拒绝策略
                rejectionStrategy.RejectionStrategy();

            TASK_QUEUE.addLast(runnable);
            TASK_QUEUE.notifyAll();//唤醒所有等待着的线程

        }

    }


    /**
     * 关闭线程
     */
    public void shutdown() throws InterruptedException {

        //如果任务队列不是空的 就等一会
        while (!TASK_QUEUE.isEmpty()) {
            Thread.sleep(50);
        }

        //此时任务已经运行完毕

        synchronized (THREAD_QUEUE){
            //判断线程的个数
            int intVal = THREAD_QUEUE.size();
            while (intVal > 0) {
                for (WorkerTask workerTask : THREAD_QUEUE) {
                    if (workerTask.getTaskState() == TaskState.BLOCKED) {//如果这个线程是被锁定的
                        workerTask.interrupt();//打断线程
                        workerTask.close();//关闭线程
                        intVal--;
                    } else {
                        try {
                            Thread.sleep(10);
                        } catch (InterruptedException e) {
                            e.printStackTrace();
                        }
                    }
                }
            }

            this.destroy = true;
            Optional.of("线程池进行关闭").ifPresent(System.out::println);
        }


    }


    private void createWorkerTask() {

        WorkerTask workerTask = new WorkerTask(GROUP, THREAD_PREFIX + sqe++);
        workerTask.start();
        THREAD_QUEUE.add(workerTask);

    }


    private static class WorkerTask extends Thread {

        private volatile TaskState taskState = TaskState.FREE;

        public TaskState getTaskState() {
            return this.taskState;
        }

        public WorkerTask(ThreadGroup threadGroup, String name) {
            super(threadGroup, name);
        }

        public void run() {
            Runnable runnable;
            //先去获取任务
            OUTER:
            while (this.taskState != TaskState.DEAD) {//如果我这个线程没有死亡

                synchronized (TASK_QUEUE) {

                    while (TASK_QUEUE.isEmpty()) {

                        try {

                            this.taskState = TaskState.BLOCKED;//线程状态改成被锁定

                            TASK_QUEUE.wait();//等待

                        } catch (InterruptedException e) {

                            break OUTER;//如果线程被打断就会出到指定的 OUTER 位置

                        }

                    }

                    runnable = TASK_QUEUE.removeFirst();//从删除队列最前的一任务   并返回

                }

                if (runnable != null) {
                    this.taskState = TaskState.RUNNING;//线程状态改成执行状态
                    runnable.run();
                    this.taskState = TaskState.FREE;//运行完成后就变成最原始状态了
                }

            }


        }

        public void close() {
            this.taskState = TaskState.DEAD;
        }

    }


    public static void main(String[] args) throws IllegalAccessException, InterruptedException {

        /**
         * 第三个参数是拒绝策略  这里又使用了 策略模式
         */
        SimpleThreadPool simpleThreadPool = new SimpleThreadPool(4, 8, 12, 2000, SimpleThreadPool.DEFAULT_REJECTION_STRATEGY);

        //SimpleThreadPool simpleThreadPool = new SimpleThreadPool();

        IntStream.rangeClosed(0, 100).forEach(i -> {


            try {

                simpleThreadPool.submit(() -> {

                    System.out.println(Thread.currentThread().getName() + ":" + i + "start");

                    try {
                        Thread.sleep(1_000);
                    } catch (InterruptedException e) {
                        e.printStackTrace();
                    }

                    System.out.println(Thread.currentThread().getName() + ":" + i + "end");

                });

            } catch (IllegalAccessException e) {
                e.printStackTrace();
            }


        });


        try {
            Thread.sleep(1000);
        } catch (InterruptedException e) {
            e.printStackTrace();
        }
        simpleThreadPool.shutdown();
        //simpleThreadPool.submit(() -> System.out.println("关闭线程池后在次提交任务。"));

    }


    public int getSize() {
        return size;
    }

    public int getQueueSize() {
        return queueSize;
    }

    public boolean isDestroy() {
        return destroy;
    }


}
